package raft

//
// this is an outline of the API that raft must expose to
// the service (or tester). see comments below for
// each of these functions for more details.
//
// rf = Make(...)
//   create a new Raft server.
// rf.Start(command interface{}) (index, Term, isleader)
//   start agreement on a new log entry
// rf.GetState() (Term, isLeader)
//   ask a Raft for its current Term, and whether it thinks it is leader
// ApplyMsg
//   each time a new entry is committed to the log, each Raft peer
//   should send an ApplyMsg to the service (or tester)
//   in the same server.
//

import (
	"context"
	"fmt"
	"math/rand"
	"sync"
	"time"
)
import "labrpc"

// import "bytes"
// import "labgob"

const(
	MaxLockDuration = time.Duration(2)*time.Millisecond
	// 超时基数 300，超时延迟为 300~400
	ElectionTimeout = 300
	RPCTimeout = 30*time.Millisecond
	// Leader 发送心跳时间间隔
	HeartBeatInterval = 150 * time.Millisecond
)

//
// as each Raft peer becomes aware that successive log entries are
// committed, the peer should send an ApplyMsg to the service (or
// tester) on the same server, via the applyCh passed to Make(). set
// CommandValid to true to indicate that the ApplyMsg contains a newly
// committed log entry.
//
// in Lab 3 you'll want to send other kinds of messages (e.g.,
// snapshots) on the applyCh; at that point you can add fields to
// ApplyMsg, but set CommandValid to false for these other uses.
//
type ApplyMsg struct {
	CommandValid bool
	Command      interface{}
	CommandIndex int
}
type Entry struct{

}

//
// A Go object implementing a single Raft peer.
//
type Raft struct {
	mu        sync.Mutex          // Lock to protect shared access to this peer's state
	peers     []*labrpc.ClientEnd // RPC end points of all peers
	persister *Persister          // Object to hold this peer's persisted state
	me        int                 // this peer's index into peers[]

	// Your data here (2A, 2B, 2C).
	// Look at the paper's Figure 2 for a description of what
	// state a Raft server must maintain.

	//configure as <raft extended> figure2
	currentTerm int
	//map[Term]votefor
	votedFor map[int]int
	log []Entry

	commitIndex int
	lastApplied int
	nextIndex []int
	matchIndex []int


	timeContext *TimeContext
	lockName string
	lockTick time.Time
	//
	isLeader bool
	state ServerStatus

	//time about
	heart *time.Timer
	elec_timer *time.Timer
	election_timeout time.Duration
	heart_tick time.Duration
	heart_timer_initial bool
	heart_begin time.Time


	preemptedCh chan bool
	timeReset bool

	lockChan chan string
	unlockChan chan string
}



// return currentTerm and whether this server
// believes it is the leader.
func (rf *Raft) GetState() (int, bool) {

	var term int
	var isleader bool
	// Your code here (2A).
	rf.mu.Lock()
	term = rf.currentTerm
	isleader = rf.state==Leader
	rf.mu.Unlock()


	return term, isleader
}
/**
wrap concurrency control machanism
*/
func (rf *Raft) lock(lockName string){
	rf.mu.Lock()
	rf.lockName = lockName
	rf.lockTick = time.Now()
	go func(){rf.lockChan<-lockName}()
	//fmt.Printf("[%v][%d]lock:%s!\n",getCurTimeStamp(),rf.me,rf.lockName)

}
func (rf *Raft) unlock(lockName string){
	if(time.Since(rf.lockTick)>MaxLockDuration){
		fmt.Printf("[%d]lock:%s too much time!\n",rf.me,rf.lockName)
		rf.Kill()
	}

	//fmt.Printf("[%v][%d]unlock:%s\n",getCurTimeStamp(),rf.me,rf.lockName)
	rf.lockName = ""

	rf.mu.Unlock()
	go func(){rf.unlockChan<-lockName}()
}
func (rf *Raft)toString() string{
	rf.lock("Reflection")
	s :=fmt.Sprintf("Reflection:[%v-%d]{Term:%d,State:%s,LockName:%s}",getCurTimeStamp(),rf.me,rf.currentTerm,rf.state,rf.lockName)
	rf.unlock("Reflection")
	return s
}

//
// save Raft's persistent state to stable storage,
// where it can later be retrieved after a crash and restart.
// see paper's Figure 2 for a description of what should be persistent.
//
func (rf *Raft) persist() {
	// Your code here (2C).
	// Example:
	// w := new(bytes.Buffer)
	// e := labgob.NewEncoder(w)
	// e.Encode(rf.xxx)
	// e.Encode(rf.yyy)
	// data := w.Bytes()
	// rf.persister.SaveRaftState(data)
}


//
// restore previously persisted state.
//
func (rf *Raft) readPersist(data []byte) {
	if data == nil || len(data) < 1 { // bootstrap without any state?
		return
	}
	// Your code here (2C).
	// Example:
	// r := bytes.NewBuffer(data)
	// d := labgob.NewDecoder(r)
	// var xxx
	// var yyy
	// if d.Decode(&xxx) != nil ||
	//    d.Decode(&yyy) != nil {
	//   error...
	// } else {
	//   rf.xxx = xxx
	//   rf.yyy = yyy
	// }
}


type AppendEntryArgs struct {
	// Your data here (2A, 2B).
	Term         int
	CandidatId   int
	LastLogIndex int
	LastLogTerm  int
}

type AppendEntryReply struct {
	// Your data here (2A).
	Term        int
	VoteGranted bool
}

//
// example RequestVote RPC arguments structure.
// field names must start with capital letters!
//
type RequestVoteArgs struct {
	// Your data here (2A, 2B).
	Term         int
	CandidatId   int
	LastLogIndex int
	LastLogTerm  int
}
// server status enumeration
type ServerStatus string
const(
	Candidate ServerStatus="Candidate"
	Follower = "Follower"
	Leader = "Leader"
)

//
// example RequestVote RPC reply structure.
// field names must start with capital letters!
//
type RequestVoteReply struct {
	// Your data here (2A).
	Term        int
	VoteGranted bool
}

//
// example RequestVote RPC handler.
//
func (rf *Raft) RequestVoteRaw(args *RequestVoteArgs, reply *RequestVoteReply) {
	// Your code here (2A, 2B).
	rf.mu.Lock()
	_,ok:=rf.votedFor[args.Term]
	var log string
	//log+=fmt.Sprintf("======On Receive RequestVote,[%d->%d]======\n",args.CandidatId,rf.me)
	log+=fmt.Sprintf("%d receive requestVote RPC,args:{ID:%d,Term:%d}\n",rf.me,args.CandidatId,args.Term)
	if ok{
		//has voted this Term,return false
		reply.Term = rf.currentTerm
		reply.VoteGranted = false
		//log+=fmt.Sprintf("%d has already voted in Term:%d!,[%d->%d],Rejected\n",rf.me,args.Term,rf.me,voted_id)
	}else {
		if args.Term > rf.currentTerm {
			//log+=fmt.Sprintf("Encounter a higher Term,become a follower!\n")
			rf.state = Follower
			rf.currentTerm = args.Term
			reply.Term = rf.currentTerm
			reply.VoteGranted = true
			rf.votedFor[args.Term] = args.CandidatId
		} else if args.Term < rf.currentTerm {
			//log+=fmt.Sprintf("Encounter a lower Term!Rejected\n")
			reply.Term = rf.currentTerm
			reply.VoteGranted = false
		} else {
			//the same Term
			reply.Term = rf.currentTerm
			reply.VoteGranted = true
			rf.votedFor[args.Term] = args.CandidatId
			//log+=fmt.Sprintf("Sent a Vote[%d->%d]\n",rf.me,args.CandidatId)
		}
	}
	reply.Term = rf.currentTerm
	rf.mu.Unlock()
	log+=fmt.Sprintf("======RequestVote Finished,[%d->%d],arg:{Candi:%d,Term:%d}Result:{Term:%d,VoteGranted:%v}======\n",args.CandidatId,rf.me,args.CandidatId,args.Term,reply.Term,reply.VoteGranted)
	fmt.Println(log)


	return
}

func (rf *Raft) RequestVote(args *RequestVoteArgs, reply *RequestVoteReply) {
	// Your code here (2A, 2B).
	rf.lock("RequestVote RPC")
	_,ok:=rf.votedFor[args.Term]
	var log string
	reply.Term = rf.currentTerm
	if ok{
		reply.VoteGranted = false	//has voted this Term,return false
	}else {
		if args.Term > rf.currentTerm {
			rf.currentTerm = args.Term
			rf.changeRole(Follower)
			reply.VoteGranted = true
			rf.votedFor[args.Term] = args.CandidatId
		} else if args.Term < rf.currentTerm {
			reply.VoteGranted = false
		} else {
			reply.VoteGranted = true
			rf.votedFor[args.Term] = args.CandidatId
		}
	}
	log+=fmt.Sprintf("======RequestVote Finished,[%d->%d],arg:{Candi:%d,Term:%d}Result:{Term:%d,VoteGranted:%v}======\n",args.CandidatId,rf.me,args.CandidatId,args.Term,reply.Term,reply.VoteGranted)
	//fmt.Println(log)
	rf.unlock("RequestVote RPC")
	return
}
//
// example RequestVote RPC handler.
//
func (rf *Raft) AppendEntry(args *AppendEntryArgs, reply *AppendEntryReply) {
	// Your code here (2A, 2B).
	rf.lock("AppendEntry")
	defer rf.unlock("AppendEntry")
	//voted_id,ok:=rf.votedFor[args.Term]
	//var log string
	//log+=fmt.Sprintf("======On Receive RequestVote,[%d->%d]======\n",args.CandidatId,rf.me)
	//log+=fmt.Sprintf("%d receive requestVote RPC,args:{ID:%d,Term:%d}\n",rf.me,args.CandidatId,args.Term)
	//if ok{
	//	//has voted this Term,return false
	//	reply.VoteGranted = false
	//	log+=fmt.Sprintf("%d has already voted in Term:%d!,[%d->%d],Rejected\n",rf.me,args.Term,rf.me,voted_id)
	//}else {
	//	if args.Term > rf.currentTerm {
	//		log+=fmt.Sprintf("Encounter a higher Term!\n")
	//		reply.VoteGranted = true
	//		rf.votedFor[args.Term] = args.CandidatId
	//	} else if args.Term < rf.currentTerm {
	//		log+=fmt.Sprintf("Encounter a lower Term!Rejected\n")
	//		reply.VoteGranted = false
	//	} else {
	//		//the same Term
	//		reply.VoteGranted = true
	//		rf.votedFor[args.Term] = args.CandidatId
	//		log+=fmt.Sprintf("Sent a Vote[%d->%d]\n",rf.me,args.CandidatId)
	//	}
	//}
	//refuse stale leader
	reply.Term = rf.currentTerm

	if args.Term<rf.currentTerm{
		return
	}
	if args.Term>rf.currentTerm{
		rf.currentTerm = args.Term
		rf.changeRole(Follower)
	}else{
		//rf.resetElectionTimer()
	}

	//if args.CandidatId==rf.me{
	//	Log(DebugL,"[%v][Stop One Self From New Election],{ID:%d,Term:%d}\n",getCurTimeStamp(),rf.me,rf.currentTerm)
	//}


	//log+=fmt.Sprintf("======Reset Election Timer,[%d->%d],Result:%v======\n",args.CandidatId,rf.me,ok)
	//fmt.Println(log)

	//reply.Term = rf.currentTerm
	return
}

//
// example code to send a RequestVote RPC to a server.
// server is the index of the target server in rf.peers[].
// expects RPC arguments in args.
// fills in *reply with RPC reply, so caller should
// pass &reply.
// the types of the args and reply passed to Call() must be
// the same as the types of the arguments declared in the
// handler function (including whether they are pointers).
//
// The labrpc package simulates a lossy network, in which servers
// may be unreachable, and in which requests and replies may be lost.
// Call() sends a request and waits for a reply. If a reply arrives
// within a timeout interval, Call() returns true; otherwise
// Call() returns false. Thus Call() may not return for a while.
// A false return can be caused by a dead server, a live server that
// can't be reached, a lost request, or a lost reply.
//
// Call() is guaranteed to return (perhaps after a delay) *except* if the
// handler function on the server side does not return.  Thus there
// is no need to implement your own timeouts around Call().
//
// look at the comments in ../labrpc/labrpc.go for more details.
//
// if you're having trouble getting RPC to work, check that you've
// capitalized all field names in structs passed over RPC, and
// that the caller passes the address of the reply struct with &, not
// the struct itself.
//
func (rf *Raft) sendRequestVote(server int, args *RequestVoteArgs, reply *RequestVoteReply) bool {
	//fmt.Printf("Send Request:peers,%d,server:%d\n",len(rf.peers),server)
	ok := rf.peers[server].Call("Raft.RequestVote", args, reply)
	return ok
}

func (rf *Raft) sendAppendEntry(server int, args *AppendEntryArgs, reply *AppendEntryReply) bool {
	//fmt.Printf("Send Request:peers,%d,server:%d\n",len(rf.peers),server)
	ok := rf.peers[server].Call("Raft.AppendEntry", args, reply)
	return ok
}


//
// the service using Raft (e.g. a k/v server) wants to start
// agreement on the next command to be appended to Raft's log. if this
// server isn't the leader, returns false. otherwise start the
// agreement and return immediately. there is no guarantee that this
// command will ever be committed to the Raft log, since the leader
// may fail or lose an election. even if the Raft instance has been killed,
// this function should return gracefully.
//
// the first return value is the index that the command will appear at
// if it's ever committed. the second return value is the current
// Term. the third return value is true if this server believes it is
// the leader.
//
func (rf *Raft) Start(command interface{}) (int, int, bool) {
	index := -1
	term := -1
	isLeader := true

	// Your code here (2B).


	return index, term, isLeader
}

//
// the tester calls Kill() when a Raft instance won't
// be needed again. you are not required to do anything
// in Kill(), but it might be convenient to (for example)
// turn off debug output from this instance.
//
func (rf *Raft) Kill() {
	// Your code here, if desired.
}
func getCurTimeStamp() string {
	timeStamp:=time.Now().Format("2006-01-02 15:04:05.000")
	return timeStamp
}
func (rf *Raft)resetElectionTimer(){
	rf.timeContext.elec_timer.Stop()
	rf.timeContext.elec_timer.Reset(rf.timeContext.elec_interval)
}

func (rf *Raft)resetHeartBeatTimer(initialFlag bool){
	rf.timeContext.heart_timer.Stop()
	if initialFlag{
		rf.timeContext.heart_timer.Reset(rf.timeContext.instant_interval)
	}else{
		rf.timeContext.heart_timer.Reset(rf.timeContext.heart_interval)
	}

}
//
// the service or tester wants to create a Raft server. the ports
// of all the Raft servers (including this one) are in peers[]. this
// server's port is peers[me]. all the servers' peers[] arrays
// have the same order. persister is a place for this server to
// save its persistent state, and also initially holds the most
// recent saved state, if any. applyCh is a channel on which the
// tester or service expects Raft to send ApplyMsg messages.
// Make() must return quickly, so it should start goroutines
// for any long-running work.
//

func Serve(rf *Raft){
	rf.lock("TimeContext")
	timeContext := &TimeContext{}
	timeContext.elec_interval = time.Duration(150+rand.Intn(150000)%100)*time.Millisecond	//election timeout 150-300 millisecond
	Log(DebugL,"[Set Election Timer],{ID:%d,tick:%d}\n",rf.me,timeContext.elec_interval)
	timeContext.elec_timer = time.NewTimer(timeContext.elec_interval)

	timeContext.heart_interval = time.Duration(100)*time.Millisecond							//lab2 assume that there is 10 heartbeats per second
	heart_beat_timer := time.NewTimer(time.Hour)
	timeContext.heart_timer = heart_beat_timer
	timeContext.instant_interval = time.Duration(0)*time.Millisecond							//trigger timer as soon as possible
	timeContext.RPCTimeout = time.Duration(30)*time.Millisecond


	timeContext.heart_initial = false

	rf.timeContext = timeContext
	rf.unlock("TimeContext")

	reflect_ticker:=time.NewTicker(time.Duration(1)*time.Millisecond)
	//RequestVote Thread
	go func(){
		for{
			select{
			case <-rf.timeContext.elec_timer.C:
				rf.lock("Election Timer")
				rf.resetElectionTimer()
				rf.unlock("Election Timer")
				go func(rf *Raft){
					RV(rf)
				}(rf)
			}
		}}()
	//AppendEntry Thread
	go func(){
		for {
			select {
			case <-rf.timeContext.heart_timer.C:
				rf.lock("Reset HeartBeat")
				rf.resetHeartBeatTimer(false)
				rf.unlock("Reset HeartBeat")
				go func(rf *Raft){
					AE(rf)
				}(rf)
				//rf.mu.Lock()
				//invalid_flag := false
				//if rf.state != Leader {
				//	invalid_flag = true
				//}
				//Log(DebugL, "[%v][SendHeartBeat],{ID:%d,Term:%d,State:%s}\n", getCurTimeStamp(), rf.me, rf.currentTerm, rf.state)
				////rf.mu.Unlock()
				//timeContext.mu.Lock()
				//if invalid_flag {
				//	timeContext.heart_timer.Stop()
				//	Log(DebugL, "[%v][Stop Heart Beat Timer],{ID:%d,Term:%d,State:%s}\n", getCurTimeStamp(), rf.me, rf.currentTerm, rf.state)
				//
				//} else {
				//	ok := timeContext.heart_timer.Reset(timeContext.heart_interval)
				//	//used like setTimeout,reset a timer after it is fired,so begin is safe in currency
				//	if ok {
				//		fmt.Printf("[%v],%d Reset Heart timer,result:%v\n", getCurTimeStamp(), rf.me, ok)
				//	}
				//	go SendHeartBeat1(rf)
				//}
				//timeContext.mu.Unlock()
				//rf.mu.Unlock()
			}
		}
	}()
	go func(rf *Raft){
		for{
			select{
			case <-reflect_ticker.C:
				fmt.Println(rf.toString())

			}
		}

	}(rf)
}


func Serve1(rf *Raft){
	for{
		elec_timeout:=time.Duration(ElectionTimeout+rand.Int63n(100))*time.Millisecond
		context,cancel := context.WithTimeout(context.Background(),elec_timeout)
		select{
		case <-context.Done():
			RV(rf)						//vote in sequential
		case <-rf.preemptedCh:			//preempted
			rf.lock("Cancel  Election Context")
			cancel()
			rf.timeReset = true
			rf.unlock("Cancel  Election Context")
		}
	}
}


//struct about time
type TimeContext struct {
	//mu sync.Mutex				//delete for code consistency
	format string
	elec_timer *time.Timer
	heart_timer *time.Timer
	elec_interval time.Duration
	heart_interval time.Duration
	RPCTimeout time.Duration

	prev_elec_tick time.Time
	prev_heart_tick time.Time
	heart_initial bool
	instant_interval time.Duration
}



type RequestVote struct{
	reply *RequestVoteReply

}
/**
function wrap status change in raft
*/
func (rf *Raft)changeRole(state ServerStatus){
	switch state {
	case Follower:
		//rawState := rf.state
		//if rawState ==Leader{
		//	rf.timeContext.heart_timer.Stop()				//if it is a leader,stop heartbeat
		//}
		rf.state = Follower
		//rf.resetElectionTimer()
		go func(){
			rf.preemptedCh<-true
		}()
	case Candidate:
		rf.state = Candidate
		rf.currentTerm = rf.currentTerm+1 					//add currentTerm By 1
		//rf.resetElectionTimer()
		rf.votedFor[rf.currentTerm]=rf.me					//vote for myself
	case Leader:

		rf.state = Leader
		//go AE(rf);											//lock will be unlocked,AE will start
		//rf.resetHeartBeatTimer(true)
		//rf.reset

	}
}
/**
RequestVote Transation
*/
func RV(rf *Raft){


	rf.lock("TransRequestVote Start")
	rf.changeRole(Candidate)							//Follower->Candidate||Candidate->Candidate
	var rawTerm = rf.currentTerm						//treat election as a transaction,rawTerm is a key variable
	var peerCount = len(rf.peers)
	rf.unlock("TransRequestVote Start")

	//time.Sleep(10 * time.Millisecond)

	var elec_mu sync.Mutex								//election mutex
	var accept = 1										//vote for myself
	var finished = 1
	var accessed=[]int{}
	var transTermFlag = false							//variable indicate TransRequestVote transaction ended
	var preemptiveFlag = false

	replyChan:=make(chan *RequestVoteReply,peerCount-1) //buffered chanenl made it nonblock
	for i,_:=range rf.peers{							//check configuration,majority,not group
		if i==rf.me{
			continue
		}
		go func(index int,rawTermArg int){
			reply:=& RequestVoteReply{}
			request:=& RequestVoteArgs{CandidatId: rf.me, Term:rawTermArg}
			if rf.sendRequestVote(index,request,reply){
				elec_mu.Lock()
				accessed = append(accessed,index)
				if !transTermFlag{						//avoid sending to a closed channel
					replyChan<-reply
				}
				elec_mu.Unlock()
			}else{

			}
		}(i,rawTerm)
	}
	//RPCTimer := time.NewTimer(RPCTimeout)
	RPCTimer := make(chan int,1)
	go func(){
		time.Sleep(RPCTimeout)
		RPCTimer<-1
	}()

	for{
		if transTermFlag{
			close(replyChan)
			break;										//break for loop
		}
		select{
		case reply:=<-replyChan:
			elec_mu.Lock()
			preemptiveFlag = preemptivHandler(rf,reply,rawTerm)
			if(preemptiveFlag){
				transTermFlag = true
				break;
			}

									//judge valid reply
			finished+=1
			if finished==peerCount{					//if all request has finished,abort wait
				transTermFlag = true
			}
			if reply.VoteGranted{accept+=1}
			if accept>=peerCount/2+1{
				transTermFlag = true
			}
			elec_mu.Unlock()
		case <-RPCTimer:							//election transaction timeout
			elec_mu.Lock()
			transTermFlag = true
			elec_mu.Unlock()
		}
	}
	//close(replyChan)
	if preemptiveFlag{
		return
	}
	if accept>=peerCount/2+1{
		rf.lock("TransRequestVote Reply 2")
		//rf.changeRole(Leader)
		rf.state = Leader
		elec_mu.Lock()
		fmt.Printf("[%v][VoteSum]{ID:%dTerm:%d},Result:%v,Accessed:%v\n",getCurTimeStamp(),rf.me,rf.currentTerm,rf.state==Leader,accessed)
		elec_mu.Unlock()
		rf.unlock("TransRequestVote Reply 2")
		go AEWrapper(rf)
	}else{
		rf.lock("TransRequestVote Reply 3")
		elec_mu.Lock()
		fmt.Printf("[%v][VoteSum]{ID:%dTerm:%d},Result:%v,Accessed:%v\n",getCurTimeStamp(),rf.me,rf.currentTerm,rf.state==Leader,accessed)
		elec_mu.Unlock()
		rf.unlock("TransRequestVote Reply 3")
	}

}
/**
Preemptive Handler
*/
func preemptivHandler(rf *Raft,reply *RequestVoteReply,rawTerm int) bool{
	preemtiveFlag := false
	rf.lock("PreemptiveHandler")
	if rf.currentTerm!=rawTerm{
		preemtiveFlag=true						//preempted by peer in a passive way
	}
	if reply.Term>rf.currentTerm{				//preempted by peer in a positive way
		rf.currentTerm = reply.Term
		preemtiveFlag = true
	}
	if preemtiveFlag{
		rf.changeRole(Follower)
	}
	rf.unlock("PreemptiveHandler")
	return preemtiveFlag
}
func AEWrapper(rf *Raft){
	for{
		rf.lock("AEWrapper Start")
		if rf.state!=Leader{
			return
			rf.unlock("AEWrapper Start")
		}
		rf.unlock("AEWrapper")

		AE(rf)

		rf.lock("AEWrapper End")
		if rf.state!=Leader{
			return
			rf.unlock("AEWrapper")
		}

		rf.unlock("AEWrapper End")
		time.Sleep(HeartBeatInterval)
	}

}


func AE(rf *Raft){

	rf.lock("AE Start")
	var peerCount = len(rf.peers)
	var rawTerm  = rf.currentTerm
	rf.unlock("AE Start")

	var ae_mu sync.Mutex
	var finished = 0
	var transTermFlag = false
	var invalidFlag = false
	var accessed=[]int{}

	request:=&  AppendEntryArgs{CandidatId: rf.me, Term:rawTerm}
	replyChan:=make(chan *AppendEntryReply,peerCount)  //buffered chanenl made it nonblock
	for i,_:=range rf.peers{							//check configuration,majority,not group
		go func(index int){
			reply:=& AppendEntryReply{}
			rf.sendAppendEntry(index,request,reply)
			if rf.sendAppendEntry(index,request,reply){
				ae_mu.Lock()
				accessed = append(accessed,index)
				ae_mu.Unlock()
				replyChan<-reply 						//decouple
			}
		}(i)
	}
	RPCTimer := time.NewTimer(time.Duration(10)*time.Millisecond)

	for{
		if transTermFlag{
			break;										//break for loop
		}
		select{
		case reply:=<-replyChan:

			rf.lock("AE Reply 1")
			if rf.currentTerm!=rawTerm{
				rf.changeRole(Follower)					//change role to follower
				transTermFlag = true
				invalidFlag = true
				rf.unlock("AE Reply 1")
				break;
			}
			if reply.Term>rf.currentTerm{			//if reply term > rf.curTerm,update rf self,abort
				rf.currentTerm = reply.Term
				rf.changeRole(Follower)
				transTermFlag = true
				invalidFlag = true
				rf.unlock("AE Reply 1")
				break;
			}
			rf.unlock("AE Reply 2")
			ae_mu.Lock()							//judge valid reply
			finished+=1
			if finished==peerCount{					//if all request has finished,abort wait
				transTermFlag = true
			}
			ae_mu.Unlock()
		case <-RPCTimer.C:							//election transaction timeout
			ae_mu.Lock()
			transTermFlag = true
			ae_mu.Unlock()
		}
	}
	//close(replyChan)
	//treat detail
	if invalidFlag{
		return
	}
	rf.lock("AE Reply 2")
	ae_mu.Lock()
	fmt.Printf("[%v][Actual SendHeartBeat],{ID:%d,Term:%d},Accessed:%v\n",getCurTimeStamp(),rf.me,rf.currentTerm,accessed)
	ae_mu.Unlock()
	rf.unlock("AE Reply 2")
}


func SendHeartBeat(rf *Raft){
	rf.mu.Lock()
	rawTerm := rf.currentTerm
	Log(DebugL,"[%v][Actual SendHeartBeat],{ID:%d,Term:%d}\n",getCurTimeStamp(),rf.me,rf.currentTerm)
	rf.mu.Unlock()
	for i,_:=range rf.peers{
		go func(index int,rawTerm int){
			reply:=& AppendEntryReply{}
			// increase Term by 1
			request:=&  AppendEntryArgs{CandidatId: rf.me, Term:rf.currentTerm}
			//index should be peers
			//log+=fmt.Sprintf("AppendEntry RPC Sent Start:[%d->%d]\n",rf.me,index)
			rf.sendAppendEntry(index,request,reply)

			if reply.Term>rf.currentTerm{
				rf.mu.Lock()
				rf.currentTerm = reply.Term
				rf.state = Follower
				rf.mu.Unlock()
			}
			//if ok{
			//	if reply.VoteGranted == true{
			//		accept+=1
			//	}
			//}else{
			//	log+=fmt.Sprintf("Request Vote Failed:[%d->%d]\n",rf.me,index)
			//}
		}(i,rawTerm)
	}
}

//
// the service or tester wants to create a Raft server. the ports
// of all the Raft servers (including this one) are in peers[]. this
// server's port is peers[me]. all the servers' peers[] arrays
// have the same order. persister is a place for this server to
// save its persistent state, and also initially holds the most
// recent saved state, if any. applyCh is a channel on which the
// tester or service expects Raft to send ApplyMsg messages.
// Make() must return quickly, so it should start goroutines
// for any long-running work.
//
func Make(peers []*labrpc.ClientEnd, me int,
	persister *Persister, applyCh chan ApplyMsg) *Raft {
	rf := &Raft{}
	rf.peers = peers
	rf.persister = persister
	rf.me = me
	// Your initialization code here (2A, 2B, 2C).
	rf.preemptedCh = make(chan bool,1)
	rf.votedFor=make(map[int]int)
	rf.lockChan = make(chan string,1)
	rf.unlockChan = make(chan string,1)
	go Serve1(rf)

	//Lock Detect
	go func(){
		lock_timeout := make(chan string,1)
		lock_finished:=false
		for{
			lockName := ""
			select {
			case lockName = <-rf.lockChan:
				//fmt.Printf("----------------------------------------------Start Lock:%s\n",lockName)
				go func(lockNameArg string) {
					time.Sleep(ElectionTimeout*2)
					lock_timeout <- lockNameArg
				}(lockName)
				lock_finished = false

			case <-rf.unlockChan:
				lock_finished = true
			case timeoutLock:=<-lock_timeout:
				if !lock_finished {
					fmt.Printf("[Raft %d],%s timeout\n", rf.me, timeoutLock)
				}
			}
		}
	}()


	// initialize from state persisted before a crash
	rf.readPersist(persister.ReadRaftState())
	return rf
}











